package com.vteba.shards.schema.service;

import java.util.List;

import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.vteba.shards.schema.bean.JsonBean;
import com.vteba.shards.schema.model.SchemaInfo;
import com.vteba.utils.charstr.Char;
import com.vteba.utils.charstr.StringUtils;
import com.vteba.utils.common.PropUtils;
import com.vteba.utils.json.FastJsonUtils;

/**
 * schema新增，删除，以及该应用的schema信息的同步。
 * @author yinlei
 * @date 2014年11月25日 上午10:02:37
 */
public class SchemaTask implements Runnable {
	private static final Logger LOGGER = LogManager.getLogger(SchemaTask.class);
	
	private ZkClient zkClient;
	private SchemaInfoService schemaInfoService;
	
	public SchemaTask(ZkClient zkClient, SchemaInfoService schemaInfoService) {
		super();
		this.zkClient = zkClient;
		this.schemaInfoService = schemaInfoService;
	}

	@Override
	public void run() {
		String appNodePath = PropUtils.get("zookeeper.app.path");
		if (StringUtils.isEmpty(appNodePath)) {
			LOGGER.warn("cannot find item[zookeeper.app.path] in config.properties.");
			return;
		}
		
		initAppSchema(appNodePath);
		
		// 订阅（监听）该app的schema的变化，添加或者移除schema
		zkClient.subscribeDataChanges(appNodePath, new IZkDataListener() {

			public void handleDataDeleted(String dataPath) throws Exception {
				LOGGER.info("deleted dataNode[{}].", dataPath);
			}

			public void handleDataChange(String dataPath, Object data) throws Exception {
				String json = "";
				if (data instanceof String) {
					json = (String) data;
				} else if (data instanceof byte[]) {
					json = new String((byte[]) data, Char.UTF8);
				} else {
					LOGGER.warn("received data is not String or byte[] format, will discard.");
				}
				if (LOGGER.isInfoEnabled()) {
					LOGGER.info("received data is[{}]", json);
				}
				JsonBean bean = FastJsonUtils.fromJson(json, JsonBean.class);
				SchemaInfo schemaInfo = bean.getSchemaInfo();
				int code = bean.getCode();
				if (code == 1) { // 新建
					if (LOGGER.isInfoEnabled()) {
						LOGGER.info("will create schema[{}].", schemaInfo.getSchemaName());
					}
					boolean result = schemaInfoService.createSchema(schemaInfo);
					if (result) { // 新增成功，将其加到SchemaHolder
						SchemaHolder.add(schemaInfo.getSchemaName(), schemaInfo);
					}
				} else if (code == -1) { // 删除
					if (LOGGER.isInfoEnabled()) {
						LOGGER.info("will remove schema[{}].", schemaInfo.getSchemaName());
					}
					boolean result = schemaInfoService.removeSchema(schemaInfo);
					if (result) { // 删除成功，将其加到SchemaHolder
						SchemaHolder.remove(schemaInfo.getSchemaName());
					}
				} else {
					LOGGER.warn("Unknown JsonBean's code[{}].", code);
				}
			}
		});
		
		// 订阅（监听）子节点的变化，同步现在可用的schema
		// 通过上面的也是可以做到的，这里是可以监听其他方式（例如命令行）对zk树的更改
		zkClient.subscribeChildChanges(appNodePath, new IZkChildListener() {

			@Override
			public void handleChildChange(String parentPath, List<String> currentChilds)
					throws Exception {
				initAppSchema(parentPath);
			}
			
		});
	}

	/**
	 * 初始化该应用所连接的schema
	 * @param appNodePath 该应用的zk节点路径
	 */
	private void initAppSchema(String appNodePath) {
		List<String> childrenList = zkClient.getChildren(appNodePath);
		if (childrenList != null) {
			// 初始化该应用所对应的数据库schema
			for (String node : childrenList) {
				String schemaJson = zkClient.readData(appNodePath + "/" + node, true);
				if (StringUtils.isEmpty(schemaJson)) {
					continue;
				}
				SchemaInfo schemaInfo = FastJsonUtils.fromJson(schemaJson, SchemaInfo.class);
				SchemaHolder.add(schemaInfo.getSchemaName(), schemaInfo);
			}
		}
	}

	public void setZkClient(ZkClient zkClient) {
		this.zkClient = zkClient;
	}

	public void setSchemaInfoService(SchemaInfoService schemaInfoService) {
		this.schemaInfoService = schemaInfoService;
	}

}
